﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ComponentModel.Composition;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Thrift;
using Thrift.Protocol;
using Thrift.Server;
using Thrift.Transport;

namespace Dubbo.Rpc.Thrift
{
    public class ThriftInvoker<T> : AbstractInvoker<T>
    {
        private URL url;
        private BlockingCollection<TSocket> sockets;
        private ConcurrentBag<IInvoker> invokers;
        private IProxyFactory proxyFactory;

        public ThriftInvoker(IProxyFactory proxyFactory, URL url, TSocket[] sockets, ConcurrentBag<IInvoker> invokers) : base(url, new string[0])
        {
            this.proxyFactory = proxyFactory;
            this.url = url;
            this.sockets = new BlockingCollection<TSocket>(new ConcurrentQueue<TSocket>(sockets));
            this.invokers = invokers;
        }

        protected override IResult DoInvoke(IInvocation<T> invocation)
        {
            TSocket transport;
            T client;

            string typeName = typeof(T).FullName;
            if (typeName.EndsWith(ThriftProtocol.IFACE))
            {
                if (sockets.TryTake(out transport, 1000))
                {
                    try
                    {
                        string clientClsName = typeName.Substring(0, typeName.IndexOf(ThriftProtocol.IFACE)) + ThriftProtocol.CLIENT;
                        Type type = Type.GetType(clientClsName + "," + typeof(T).Assembly.FullName);
                        ConstructorInfo ctr = type.GetConstructor(new Type[] { typeof(TProtocol) });

                        TProtocol protocol = new TBinaryProtocol(transport);
                        client = (T)ctr.Invoke(new object[] { protocol });
                        if (!transport.IsOpen)
                        {
                            transport.Open();
                        }
                        IInvoker<T> tagert = proxyFactory.GetInvoker(client, url);
                        return tagert.Invoke(invocation);
                    }
                    catch (Exception e)
                    {
                        //logger.error(e.getMessage(), e);
                        throw new RpcException("Fail to invoke method: " + invocation.MethodName + " on service(" + url + "): " + e.Message, e);
                    }
                    finally
                    {
                        sockets.Add(transport);
                    }
                }
                else
                {
                    throw new RpcException("Timeout to get TSocket for service(" + url + "): ");
                }
            }
            else
            {
                throw new RpcException("Fail to create remoting client for service(" + url + "): " + typeName);
            }
        }
        
        public override bool IsAvailable
        {
            get
                {
                //if (!base.IsAvailable)
                //{
                //    return false;
                //}

                //foreach (var socket in sockets)
                //{
                //    if (socket.IsOpen)
                //    {
                //        return true;
                //    }
                //}
                //return false;
                return true;
            }
            
        }

        private object obj = new object();

        public override void Destroy()
        {
            lock (obj)
            {
                if (base.IsDestroyed)
                {
                    return;
                }

                base.Destroy();

                if (invokers != null)
                {
                    IInvoker invoker = this;
                    invokers.TryTake(out invoker);
                }

                foreach (var socket in sockets)
                {
                    try
                    {
                        socket.Close();
                        socket.Dispose();
                    }
                    catch (Exception e)
                    {
                        //logger.warn(e.getMessage(), e);
                    }
                }
            }
        }
    }
}
